package com.motu.monstercity.server.gate;


import com.motu.monstercity.module.common.network.ResponseUtils;
import com.motu.monstercity.module.common.utility.Constant;
import com.motu.monstercity.module.common.utility.EventBusConstant;
import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Enumeration;
import java.util.concurrent.ConcurrentHashMap;

public class BroadcastVerticle extends AbstractVerticle {
    public final static Logger logger = LoggerFactory.getLogger(BroadcastVerticle.class);

    @Override
    public void start(Promise<Void> future) throws Exception {
        EventBus eb = vertx.eventBus();
        String address = EventBusConstant.getAddress(EventBusConstant.GATE_BROADCAST);
        MessageConsumer<JsonObject> consumer = eb.consumer(address);
        //logger.info("---===#*#*#*#===---* gate BroadcastVerticle start address:" + address);
        consumer.handler(message -> {
            // 踢掉服务器玩家
            if (message.headers().contains("kickserver")) {
                SocketManager.kickServerUser(new JsonArray(message.headers().get("kickserver")));
                return;
            }

            long sourceUserId = 0;
            if (message.headers().contains("suid")) {
                sourceUserId = Long.parseLong(message.headers().get("suid"));
            }
            int targetLanguage = 0;
            if (message.headers().contains("lang")) {
                targetLanguage = Integer.parseInt(message.headers().get("lang"));
            }
            JsonObject recData = message.body();
            int resCode = recData.getInteger("code");
            byte[] resData = recData.getBinary("data");
            logger.info("---===#*#*#*#===---* Received  Broadcast  resCode:"+resCode+" resData:" + resData.length);

            // 广播给指定的服务器
            if (message.headers().contains("slist")) {
                JsonArray serverList = new JsonArray(message.headers().get("slist"));
                ConcurrentHashMap<Long, ServerWebSocket> list = SocketManager.getSocketList();
                Enumeration<ServerWebSocket> e = list.elements();
                while (e.hasMoreElements()) {
                    ServerWebSocket info = e.nextElement();
                    if (sourceUserId > 0 && SocketInfo.getUserId(info) == sourceUserId) {//过滤掉自己
                        continue;
                    }
                    if (targetLanguage > 0 && targetLanguage != SocketInfo.getLanguage(info)) {
                        continue;// 只发特定语言玩家
                    }
                    if (serverList.size() > 0 && !Tool.isInList(serverList, SocketInfo.getServerId(info))) {
                        continue;
                    }
                    if (message.headers().contains("scopeType") && Integer.parseInt(message.headers().get("scopeType")) == 1
                            && !SocketInfo.isChatWorld(info)) {
                        continue;// 玩家没停在跨服聊天页面时,不进行推送
                    }
                    ResponseUtils.push(info, resCode, resData);
                }
            }

            // 广播给指定的userid
            if (message.headers().contains("ulist")) {
                JsonArray userList = new JsonArray(message.headers().get("ulist"));
                ConcurrentHashMap<Long, ServerWebSocket> list = SocketManager.getSocketList();
                for (int i = 0; i < userList.size(); i++) {
                    long userId = userList.getLong(i);
                    if (list.containsKey(userId)) {
                        ServerWebSocket info = list.get(userId);
                        ResponseUtils.push(info, resCode, resData);
                    }
                }
            }

            // 广播给所有人
            if (message.headers().contains("all")) {
                boolean checkServerId = message.headers().contains("stype") && message.headers().contains("serverids");
                boolean checkPlatform = message.headers().contains("ptype") && message.headers().contains("platforms");
                byte stype = message.headers().contains("stype") ? Byte.parseByte(message.headers().get("stype")) : 0;
                String serverids = message.headers().get("serverids");
                byte ptype = message.headers().contains("ptype") ? Byte.parseByte(message.headers().get("ptype")) : 0;
                String platforms = message.headers().get("platforms");
                checkServerId = checkServerId && stype != Constant.ACT_RTYPE_ALL;
                checkPlatform = checkPlatform && ptype != Constant.ACT_RTYPE_ALL;
                ConcurrentHashMap<Long, ServerWebSocket> list = SocketManager.getSocketList();
                for (ServerWebSocket socket : list.values()) {
                    if (socket.isClosed()) {
                        continue;
                    }
                    // 判断是否在服务器范围内
                    if (checkServerId) {
                        int serverid = SocketInfo.getServerId(socket);
                        if (!isServerRange(serverid, stype, serverids)) {
                            continue;
                        }
                    }
                    // 判断是否在渠道范围内
                    if (checkPlatform) {
                        int platform = SocketInfo.getPlatform(socket);
                        if (!isPlatformRange(platform, ptype, platforms)) {
                            continue;
                        }
                    }
                    ResponseUtils.push(socket, resCode, resData);
                }
            }
        });

        consumer.completionHandler(res2 -> {
            if (res2.succeeded()) {
                future.complete();
                logger.info("The handler registration has reached all nodes");
            } else {
                future.fail("The handler registration has failed");
                logger.error("Registration failed! " + res2.cause().toString());
            }
        });
    }

    /***
     * 是否在平台范围内
     * @param userPlatform 用户平台
     */
    public boolean isPlatformRange(int userPlatform, byte ptype, String platform) {
        if (ptype == Constant.ACT_RTYPE_ALL) {
            return true;
        } else if (ptype == Constant.ACT_RTYPE_IN) {
            JsonArray jsonArray = new JsonArray(platform);
            if (jsonArray.contains(userPlatform)) {
                return true;
            }
        } else if (ptype == Constant.ACT_RTYPE_NOTIN) {
            JsonArray jsonArray = new JsonArray(platform);
            if (!jsonArray.contains(userPlatform)) {
                return true;
            }
        }
        return false;
    }

    /**
     * 判断是否有包含本服
     */
    public boolean isServerRange(int serverId, byte stype, String serverids) {
        boolean contain = false;// 是否包含
        if (stype == Constant.ACT_RTYPE_ALL) {
            contain = true;
        } else if (stype == Constant.ACT_RTYPE_IN) {
            JsonArray jsonArray = new JsonArray(serverids);
            if (jsonArray.contains(serverId)) {
                contain = true;
            }
        } else if (stype == Constant.ACT_RTYPE_NOTIN) {
            JsonArray jsonArray = new JsonArray(serverids);
            if (!jsonArray.contains(serverId)) {
                contain = true;
            }
        }
        return contain;
    }


}
